package cn.doitedu.rtdw.etl;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job04_video_play_olap_base {

    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建映射表，映射kafka中的dwd行为明细宽表
        // 建表，映射kafka中dwd层的行为日志明细
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,device_type string                                 "+
                        "     ,release_channel string                             "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );



        // 创建映射表，映射doris中的视频播放olap分析基础聚合表

        tenv.executeSql(
                        " create table video_olap_timelong_base_doris (               "+
                        "     dt                   date                               "+
                        "     ,session_id          varchar(200)                       "+
                        "     ,user_id             bigint                             "+
                        "     ,province            varchar(200)                       "+
                        "     ,city                varchar(200)                       "+
                        "     ,region              varchar(200)                       "+
                        "     ,video_id            int                                "+
                        "     ,play_id             varchar(200)                       "+
                        "     ,flag                int                                "+
                        "     ,video_name          varchar(200)                       "+
                        "     ,video_type          varchar(200)                       "+
                        "     ,video_album         varchar(200)                       "+
                        "     ,video_author        varchar(200)                       "+
                        "     ,video_timelong      bigint                             "+
                        "     ,create_time         timestamp(3)                       "+
                        "     ,play_start_time     bigint                             "+
                        "     ,play_end_time       bigint                             "+
                        " )  with (                                                   "+
                        "     'connector' = 'doris',                                  "+
                        "     'fenodes' = 'doitedu:8030',                             "+
                        "     'table.identifier' = 'dws.video_olap_timelong_base',    "+
                        "     'username' = 'root',                                    "+
                        "     'password' = 'root',                                    "+
                        "     'sink.label-prefix' = 'doit42_no5'                      "+
                        " )                                                           "
        );



        // 统计sql开发
        tenv.executeSql(
                " create temporary view time_long_res AS                                                                                                  "+
                " WITH tmp1 AS (                                                                                                    "+
                        "     SELECT                                                                                                        "+
                        "          *,                                                                                                       "+
                        "     	 cast(properties['video_id'] as int) as video_id,                                                                        "+
                        "     	 properties['play_id'] as play_id                                                                           "+
                        "     from dwd_events_kafka                                                                                         "+
                        "     where event_id in ('video_play','video_hb','video_pause','video_resume','video_stop')                         "+
                        " ),                                                                                                                "+
                        " tmp2 as (                                                                                                         "+
                        "     SELECT                                                                                                        "+
                        "         *,                                                                                                        "+
                        "     	sum(if(event_id = 'video_resume',1,0)) over(partition by user_id,video_id,play_id order by rt) as flag      "+
                        "     FROM tmp1                                                                                                     "+
                        " )                                                                                                                 "+
                        "                                                                                                                   "+
                        " SELECT                                                                                                            "+
                        "     session_id                                                                                                    "+
                        "     ,user_id                                                                                                      "+
                        "     ,province                                                                                                     "+
                        "     ,city                                                                                                         "+
                        "     ,region                                                                                                       "+
                        "     ,video_id                                                                                                     "+
                        "     ,play_id                                                                                                      "+
                        "     ,flag                                                                                                         "+
                        "     ,proctime() as pt                                                                                             "+
                        " 	,min(action_time) as play_start_time                                                                            "+
                        " 	,max(action_time) as play_end_time                                                                              "+
                        " FROM TABLE(                                                                                                       "+
                        "     TUMBLE(TABLE tmp2,DESCRIPTOR(rt), INTERVAL '1' MINUTE )                                                       "+
                        " )                                                                                                                 "+
                        " GROUP BY                                                                                                          "+
                        "     window_start                                                                                                  "+
                        " 	,window_end                                                                                                     "+
                        "     ,session_id                                                                                                   "+
                        "     ,user_id                                                                                                      "+
                        "     ,province                                                                                                     "+
                        "     ,city                                                                                                         "+
                        "     ,region                                                                                                       "+
                        "     ,video_id                                                                                                     "+
                        "     ,play_id                                                                                                      "+
                        "     ,flag                                                                                                         "
        );


        // 创建hbase中的 视频信息维表的  映射表
        tenv.executeSql("create table dim_video_info_hbase(   " +
                "   video_id    int,                   " +
                "   f ROW<video_name string,video_type string, " +
                "video_album string,video_author string," +
                "video_timelong bigint,create_time timestamp(3)>  " +
                ") with (                              " +
                "'connector' = 'hbase-2.2',            " +
                "'table-name' = 'dim_video_info',       " +
                "'zookeeper.quorum' = 'doitedu:2181'   " +
                ")");



        // 做视频主题维度的关联(look up)
        tenv.executeSql(
                        " INSERT INTO   video_olap_timelong_base_doris  "+
                        " SELECT                            "+
                        "     to_date(date_format(to_timestamp_ltz(play_start_time,3),'yyyy-MM-dd')) as dt     "+
                        "     ,a.session_id                  "+
                        "     ,a.user_id                    "+
                        "     ,a.province                   "+
                        "     ,a.city                       "+
                        "     ,a.region                     "+
                        "     ,a.video_id                   "+
                        "     ,a.play_id                    "+
                        "     ,a.flag                       "+
                        "     ,b.f.video_name               "+
                        " 	  ,b.f.video_type               "+
                        " 	  ,b.f.video_album              "+
                        " 	  ,b.f.video_author             "+
                        " 	  ,b.f.video_timelong           "+
                        " 	  ,b.f.create_time              "+
                        "     ,a.play_start_time            "+
                        "     ,a.play_end_time              "+
                        " FROM time_long_res a              "+
                        " LEFT JOIN dim_video_info_hbase    "+
                        " FOR SYSTEM_TIME AS OF a.pt  as b  "+
                        " ON a.video_id = b.video_id	    "

        );





    }
}
